package com.test.kfk.code.consume;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author chen_lei
 * @date 2020/11/18 10:45
 */
@Component
public class KafkaConsumer {

    /**
     * @Title 指定topic、partition、offset消费
     * @Description 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
     * @Author long.yuan
     * @Date 2020/3/22 13:38
     * @Param [record]
     * @return void
     **/
//    @KafkaListener(id = "consumer1",groupId = "felix-group",topicPartitions = {
//            @TopicPartition(topic = "topic1", partitions = { "0" }),
//            @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
//    })
    //属性解释：
    // id：消费者ID；
    //
    // groupId：消费组ID；
    //
    // topics：监听的topic，可监听多个；
    //
    // topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
    //
    //上面onMessage2监听的含义：监听topic1的0号分区，同时监听topic2的0号分区和topic2的1号分区里面offset从8开始的消息。
    //
    //注意：topics和topicPartitions不能同时使用


    // 消费监听
//    @KafkaListener(id = "consumer1",groupId = "group1",topicPartitions = {
//            @TopicPartition(topic = "topic8", partitions = { "0" }),
//          //  @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
//    },
//    containerFactory = "filterContainerFactory")
//    public void onMessage1(ConsumerRecord<String, String> record){
//        // 消费的哪个topic、partition的消息,打印出消息内容
//        System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
//    }



/**********************************************************************************************************/



    public static final String usertopic ="topic-a";


    //批量消费
//    @KafkaListener(id = "manualAck3",groupId = "group2", topics = usertopic,
//            errorHandler = "consumerAwareErrorHandler",
//            containerFactory = "consumeFilter")
//    public void manualAck3(List<ConsumerRecord<?, ?>> records,Acknowledgment ack)throws Exception {
//        System.out.println("manualAckTopic1批量消费一次，records.size()=" + records.size());
//        for (ConsumerRecord<?, ?> record : records) {
//
//            System.out.println(record.key() + " | " + record.value());
//
////            Object key = record.key();
////            Object value = record.value();
////            String topic = record.topic();
////            Headers headers = record.headers();
////            long offset = record.offset();
//
//        }
//        ack.acknowledge();
//
//        //测试异常
//     //   throw new RuntimeException("异常");
//    }


    /**
     * 通过增加分区 提升消费能力
     */

    //批量消费
    @KafkaListener(id = "manualAck1",groupId = "group1",
            topicPartitions = {
                @TopicPartition(topic = usertopic, partitions = { "0" })
            },
            errorHandler = "consumerAwareErrorHandler",
            containerFactory = "consumeFactory1")
    public void manualAck1(List<ConsumerRecord<?, ?>> records,Acknowledgment ack)throws Exception {
        System.out.println("manualAck1批量消费一次，records.size()="+ records.size());

//        for (ConsumerRecord<?, ?> record : records) {
//            System.out.println(record.key() + " | " + record.value());
//        }
        ack.acknowledge();

    }

    @KafkaListener(id = "manualAck2",groupId = "group1",
            topicPartitions = {
                    @TopicPartition(topic = usertopic, partitions = { "1" })
            },
            errorHandler = "consumerAwareErrorHandler",
            containerFactory = "consumeFactory1")
    public void manualAck2(List<ConsumerRecord<?, ?>> records,Acknowledgment ack)throws Exception {
        System.out.println("manualAck2批量消费一次，records.size()="+ records.size());

//        for (ConsumerRecord<?, ?> record : records) {
//            System.out.println(record.key() + " | " + record.value());
//        }

        ack.acknowledge();
    }



}
